package com.wdl.user

import org.apache.flink.api.java.tuple.{Tuple, Tuple1}
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector


class ItemCountWindowResult() extends WindowFunction[Long, ItemViewCount, Tuple, TimeWindow] {

  override def apply(key: Tuple, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
    /** 需要调用 java 的 Tuple1 才能取到 f0 */
    val itemId: Long = key.asInstanceOf[Tuple1[Long]].f0

    /** 窗口结束时间 */
    val windowEnd: Long = window.getEnd

    /** 统计次数 */
    val count: Long = input.iterator.next()

    out.collect(ItemViewCount(itemId, windowEnd, count))
  }

}
